package main

import (
	"errors"
	"fmt"
	"os"
	"path"
	. "pedro/pbitcask/internal"
	"strings"
	"sync"
)

const (
	DEFAULT_SIZE_THRESHOLD = 2000 * 1024
)

type IndexItem struct {
	fileId        uint32
	valueSize     uint32
	valuePosition int64
	timestamp     int64
}

type Bitcask struct {
	storeDir     string // 存储目录
	activeFile   *os.File
	activeFileId uint32
	// writePos   int64 // 写位置
	olderFiles map[uint32]*os.File

	keydir map[string]IndexItem // 内存索引

	mu sync.RWMutex
}

func NewBitcask(dir string) (*Bitcask, error) {
	var (
		storeDir string
		err      error
		pwd      string
	)
	if path.IsAbs(dir) {
		storeDir = dir
	} else {
		pwd, err = os.Getwd()
		if err != nil {
			return nil, err
		}
		storeDir = path.Join(pwd, dir)
	}
	if IsFile(storeDir) {
		return nil, errors.New("input path must be dir")
	}
	if ok, _ := PathExists(storeDir); !ok {
		// 路径若是不存在，则新建该目录
		err = os.MkdirAll(storeDir, os.ModePerm)
		if err != nil {
			return nil, errors.New("can't create data dir")
		}
	}
	var (
		activeFile   *os.File
		activeFileId uint32
		files        []string
		dataFiles    []string
		hintFiles    []string
	)
	files = ScanDir(storeDir)
	for _, file := range files {
		if strings.Contains(file, "data-") {
			dataFiles = append(dataFiles, file)
		} else if strings.Contains(file, "hint-") {
			hintFiles = append(hintFiles, file)
		}
	}
	activeFileId = uint32(len(dataFiles))
	// 无论如何，都不管旧数据，直接新开一个当做新的 active file
	filename := fmt.Sprintf("data-%d.db", activeFileId)
	activeFile, err = os.Create(path.Join(storeDir, filename))
	if err != nil {
		return nil, err
	}
	cask := Bitcask{
		storeDir:     storeDir,
		activeFile:   activeFile,
		activeFileId: activeFileId,
		olderFiles:   make(map[uint32]*os.File),
		keydir:       make(map[string]IndexItem),
	}
	// 索引，如果有 hint file 则通过 hint file 来新建
	// 否则，通过扫描 data file 来新建
	// 读取 hint file 加载索引
	cask.loadKeydir(hintFiles)
	// 打开旧文件
	for _, dataFile := range dataFiles {
		id, err := GetFileId(dataFile)
		if err != nil {
			continue
		}
		file, err := os.Open(dataFile)
		if err != nil {
			continue
		}
		uid := uint32(id)
		cask.loadKeydirFromFile(file, uid) // 从 data file 中加载索引
		cask.olderFiles[uid] = file
	}
	return &cask, nil
}

func (b *Bitcask) loadKeydirFromFile(file *os.File, id uint32) error {
	var (
		sz            int
		err           error
		valuePosition int64 = 0
	)
	for {
		// 前面固定的数据
		header := make([]byte, 4+8+4+4)
		vp := valuePosition
		vp += int64(len(header))
		if err != nil {
			return err
		}
		sz, err = file.Read(header)
		if err != nil {
			return err
		}
		if sz != len(header) {
			return errors.New("read data header err")
		}
		e := Entry{}
		e.DecodeHeader(header)
		key := make([]byte, e.KeySize)
		value := make([]byte, e.ValueSize)
		sz, err = file.Read(key)
		if err != nil {
			return err
		}
		vp += int64(len(key))
		sz, err = file.Read(value)
		if err != nil {
			return err
		}
		vp += int64(len(value))
		e.SetKV(header, key, value)
		skey := string(key)
		val, ok := b.keydir[skey]
		if ok {
			// 查看 timestamp 是否更新
			if e.Timestamp > val.timestamp {
				b.keydir[skey] = IndexItem{
					fileId:        uint32(id),
					valueSize:     e.ValueSize,
					valuePosition: valuePosition,
					timestamp:     e.Timestamp,
				}
			}
		} else {
			b.keydir[skey] = IndexItem{
				fileId:        uint32(id),
				valueSize:     e.ValueSize,
				valuePosition: valuePosition,
				timestamp:     e.Timestamp,
			}
		}
		valuePosition = vp
	}
}

// 从磁盘中加载索引
func (b *Bitcask) loadKeydir(hintFiles []string) error {
	var (
		err  error
		file *os.File
		sz   int
		id   int
	)
	for _, hintFile := range hintFiles {
		id, err = GetFileId(hintFile)
		if err != nil {
			continue
		}
		file, err = os.Open(hintFile)
		if err != nil {
			return err
		}
		// 前面固定的数据
		header := make([]byte, 8+4+4+8)
		sz, err = file.Read(header)
		if err != nil {
			return err
		}
		if sz != len(header) {
			return errors.New("read hint header data err")
		}
		item := HintItem{}
		item.DecodeHeader(header)
		key := make([]byte, item.KeySize)
		sz, err = file.Read(key)
		if err != nil {
			return err
		}
		if sz != len(key) {
			return errors.New("read hint key data err")
		}
		item.Key = key
		skey := string(key)
		val, ok := b.keydir[skey]
		if ok {
			// 查看 timestamp 是否更新
			if item.Timestamp > val.timestamp {
				b.keydir[skey] = IndexItem{
					fileId:        uint32(id),
					valueSize:     item.ValueSize,
					valuePosition: item.ValuePosition,
					timestamp:     item.Timestamp,
				}
			}
		} else {
			b.keydir[skey] = IndexItem{
				fileId:        uint32(id),
				valueSize:     item.ValueSize,
				valuePosition: item.ValuePosition,
				timestamp:     item.Timestamp,
			}
		}
	}
	return nil
}

func (b *Bitcask) Put(key string, value []byte) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	var (
		sz        int
		err       error
		valuePos  int64
		valueSize = uint32(len(value))
	)
	entry := Entry{
		KeySize:   uint32(len(key)),
		ValueSize: valueSize,
		Key:       []byte(key),
		Value:     value,
	}
	data := entry.Encode()
	valuePos, err = FileSize(b.activeFile)
	if err != nil {
		return err
	}
	// 判断 valuePos，如果超过了阈值则新建 active 文件
	if int(valuePos)+len(data) > DEFAULT_SIZE_THRESHOLD {
		b.rollActiveFile()
	}
	sz, err = b.activeFile.Write(data)
	if err != nil {
		return err
	}
	if sz != len(data) {
		return errors.New("write active file err")
	}
	b.keydir[key] = IndexItem{
		fileId:        b.activeFileId,
		valueSize:     valueSize,
		valuePosition: valuePos,
		timestamp:     entry.Timestamp,
	}
	return nil
}

func (b *Bitcask) rollActiveFile() error {
	fileId := b.activeFileId + 1
	filename := fmt.Sprintf("data-%d.db", fileId)
	activeFile, err := os.Create(path.Join(b.storeDir, filename))
	if err != nil {
		return err
	}
	b.olderFiles[b.activeFileId] = b.activeFile
	b.activeFile = activeFile
	b.activeFileId += 1
	return nil
}

func (b *Bitcask) Get(key string) ([]byte, error) {
	b.mu.RLock()
	defer b.mu.RUnlock()

	item, ok := b.keydir[key]
	if !ok {
		return nil, errors.New("key not found")
	}
	var (
		valuePosition = item.valuePosition
		err           error
		sz            int
		//                  crc timestamp key_sz val_sz key value
		data = make([]byte, 4+8+4+4+len(key)+int(item.valueSize))
	)
	fileId := item.fileId
	var file *os.File
	if fileId == b.activeFileId {
		file = b.activeFile
	} else {
		file, ok = b.olderFiles[fileId]
		if !ok {
			return nil, errors.New("key not found")
		}
	}
	sz, err = file.ReadAt(data, valuePosition)
	if err != nil {
		return nil, err
	}
	if sz != len(data) {
		return nil, errors.New("read file err")
	}
	entry := Entry{}
	entry.Decode(data)
	return entry.Value, nil
}

func (b *Bitcask) Close() error {
	b.activeFile.Close()
	for _, file := range b.olderFiles {
		file.Close()
	}
	return nil
}

// 合并 older files 生成新的文件
// hint file 是在 merge 的时候生成的，但是 close 的时候
func (b *Bitcask) merge() error {
	b.mu.Lock()
	defer b.mu.Unlock() // 必须加锁

	//if b.activeFileId == 0 {
	//	return nil
	//}
	// fileId file
	// 在 storeDir 下新建 tmp 文件夹存放数据文件
	var (
		tmpDir   string
		err      error
		dataFile *os.File
		hintFile *os.File
	)
	tmpDir = os.TempDir()
	err = os.MkdirAll(tmpDir, os.ModePerm)
	if err != nil {
		return err
	}
	var mergeFiles = make(map[uint32]*os.File)
	var dataId uint32 = 0
	var hintId uint32 = 0
	var hintFiles = make(map[uint32]*os.File)
	filename := fmt.Sprintf("data-%d.db", dataId)
	hintFilename := fmt.Sprintf("hint-%d.db", hintId)
	dataFile, err = os.Create(path.Join(tmpDir, filename))
	if err != nil {
		return err
	}
	hintFile, err = os.Create(path.Join(tmpDir, hintFilename))
	if err != nil {
		return err
	}
	mergeFiles[dataId] = dataFile
	hintFiles[hintId] = hintFile
	// 根据索引来重建文件
	for key, item := range b.keydir {
		var (
			valuePosition = item.valuePosition
			err           error
			sz            int
			data          = make([]byte, 4+8+4+4+len(key)+int(item.valueSize))
			ok            bool
		)
		fileId := item.fileId
		var file *os.File
		if fileId == b.activeFileId {
			continue // active 文件不需要重建
		}
		file, ok = b.olderFiles[fileId]
		if !ok {
			continue
		}
		sz, err = file.ReadAt(data, valuePosition)
		if err != nil {
			continue
		}
		if sz != len(data) {
			continue
		}
		newValPosition, err := FileSize(dataFile)
		// 文件太大了
		if int(newValPosition)+sz > DEFAULT_SIZE_THRESHOLD {
			dataId += 1
			hintId += 1
			filename := fmt.Sprintf("data-%d.db", dataId)
			hintFilename := fmt.Sprintf("hint-%d.db", hintId)
			dataFile, err = os.Create(path.Join(tmpDir, filename))
			if err != nil {
				return err
			}
			hintFile, err = os.Create(path.Join(tmpDir, hintFilename))
			if err != nil {
				return err
			}
			mergeFiles[dataId] = dataFile
			hintFiles[hintId] = hintFile
			newValPosition = 0 // 从 0 重新开始
		}
		if err != nil {
			continue
		}
		it := IndexItem{
			fileId:        dataId,
			valueSize:     item.valueSize,
			valuePosition: newValPosition,
			timestamp:     item.timestamp,
		}
		// 写数据文件
		dataFile.Write(data)
		// 更新索引
		b.keydir[key] = it
		// 写 hint 文件
		hintIt := HintItem{
			Timestamp:     item.timestamp,
			KeySize:       uint32(len(key)),
			ValueSize:     item.valueSize,
			ValuePosition: newValPosition,
			Key:           []byte(key),
		}
		hintData := hintIt.Encode()
		hintFile.Write(hintData)
	}
	// 复制 active 文件
	b.activeFileId = dataId + 1
	activeFilename := fmt.Sprintf("data-%d.db", b.activeFileId)
	// 关闭、删除文件
	for _, file := range b.olderFiles {
		file.Sync()
		file.Close()
		os.Remove(file.Name())
	}
	// 移动
	b.activeFile.Sync()
	b.activeFile.Close()
	err = os.Rename(b.activeFile.Name(), path.Join(b.storeDir, activeFilename))
	if err != nil {
		return err
	}
	for _, file := range mergeFiles {
		file.Sync()
		file.Close()
		err = os.Rename(file.Name(), path.Join(b.storeDir, path.Base(file.Name())))
		if err != nil {
			return err
		}
	}
	for _, file := range hintFiles {
		file.Sync()
		file.Close()
		err = os.Rename(file.Name(), path.Join(b.storeDir, path.Base(file.Name())))
		if err != nil {
			return err
		}
	}
	return nil
}

func main() {
	var (
		store *Bitcask
		err   error
		name  []byte
	)
	store, err = NewBitcask("./data")
	if err != nil {
		panic(err)
	}
	fmt.Printf("%v\n", store.keydir)
	err = store.Put("age", []byte("25"))
	if err != nil {
		panic(err)
	}
	name, err = store.Get("name")
	fmt.Printf("name => %s\n", name)
	err = store.merge()
	if err != nil {
		panic(err)
	}
	//err = store.Close()
	//if err != nil {
	//	panic(err)
	//}
}
